package com.net.xpay.sync.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.common.base.Joiner;
import com.net.common.concurrent.ThreadUtil;
import com.net.common.util.StreamUtil;
import com.net.xpay.common.helper.NoticeHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;


@Slf4j
@Service
public class CanalSyncClient extends Thread {

    @Autowired
    private List<SyncEventHandler> syncEventHandlers;

    @Autowired
    private NoticeHelper noticeHelper;

    @Value("${canal.server.port}")
    private String serverPort;

    @Value("${canal.server.dest}")
    private String dest;


    private Map<Pattern, SyncEventHandler> syncEventHandlerMap;

    @Override
    public void run() {
        if (CollectionUtils.isEmpty(syncEventHandlers)) {
            log.error("no canal event handler");
            return;
        }

        List<String> subscribes = StreamUtil.map(syncEventHandlers, SyncEventHandler::getSubscribeRegex);

        syncEventHandlerMap = StreamUtil.toMap(syncEventHandlers, handler -> Pattern.compile(handler.getSubscribeRegex()));

        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                Integer.parseInt(serverPort)), dest, "canal", "D35DB127DB631E6E27C6B75E8D376B04F64FAF83");

        try {
            connector.connect();
            connector.subscribe(Joiner.on(",").join(subscribes));
            connector.rollback();
            long count = 0L;
            for (; ; ) {
                Long batchId = null;
                try {
                    //获取指定数量的数据
                    Message message = connector.getWithoutAck(32);
                    batchId = message.getId();

                    doSyncMessage(message);
                    // 提交确认
                    connector.ack(batchId);
                } catch (Exception e) {
                    log.error("canal sync error", e);
                    noticeHelper.sendSystemMonitorNotice("canal sync error", "text");
                    if (batchId != null) {
                        // 处理失败, 回滚数据
                        connector.rollback(batchId);
                    }

                    ThreadUtil.safeSleep(30000, "canal error sleep 30s");
                }

                ThreadUtil.safeSleep(50, "canal sleep");

                if (count++ % 200 == 0L) {
                    log.info("canal heart beat..");
                }
            }

        } finally {
            connector.disconnect();
        }
    }

    private void doSyncMessage(Message message) {
        long batchId = message.getId();
        int size = message.getEntries().size();
        if (batchId == -1 || size == 0) {
            return;
        }

        for (CanalEntry.Entry entry : message.getEntries()) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("parser has an error|entry={}", e, entry);
                throw new RuntimeException("parser has an error , data:" + entry.toString(),
                        e);
            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                printRowData(rowData);

                CanalSyncContext context = CanalSyncContext.getInstance(entry, rowData);

                if (eventType == CanalEntry.EventType.DELETE) {
                    //删除
                    handleDeleteEvent(context);
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    //插入
                    handleInsertEvent(context);
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    //跟新
                    handleUpdateEvent(context);
                } else {
                    log.warn("doSyncMessage unknown event type={}", eventType);
                }
            }
        }

    }

    private void handleUpdateEvent(CanalSyncContext context) {
        SyncEventHandler handler = selectHandler(context.getSchemaName(), context.getTableName());
        if (handler == null) {
            return;
        }
        handler.handleUpdateEvent(context);
    }

    private void handleInsertEvent(CanalSyncContext context) {
        SyncEventHandler handler = selectHandler(context.getSchemaName(), context.getTableName());
        if (handler == null) {
            return;
        }
        handler.handleInsertEvent(context);
    }

    private void handleDeleteEvent(CanalSyncContext context) {
        SyncEventHandler handler = selectHandler(context.getSchemaName(), context.getTableName());
        if (handler == null) {
            return;
        }
        handler.handleDeleteEvent(context);
    }

    private void printRowData(CanalEntry.RowData rowData) {
        printColumn("before", rowData.getBeforeColumnsList());
        printColumn("after", rowData.getAfterColumnsList());
    }

    private static void printColumn(String prefix, List<CanalEntry.Column> columns) {
        if (CollectionUtils.isEmpty(columns)) {
            return;
        }
        log.info(prefix + ":" + CanalBeanHelper.convertToBeanJson(columns));
    }

    private SyncEventHandler selectHandler(String schemaName, String tableName) {
        for (Map.Entry<Pattern, SyncEventHandler> entry : syncEventHandlerMap.entrySet()) {
            if (entry.getKey().matcher(schemaName + "." + tableName).matches()) {
                return entry.getValue();
            }
        }

        log.warn("selectHandler but no matched handler|schemaName={}|tableName={}", schemaName, tableName);
        noticeHelper.sendSystemInfoNotice("selectHandler but no matched handler|schemaName=" + schemaName + "|tableName=" + tableName, "text");
        return null;
    }
}
